﻿#include "threadpool.h"

#include <iostream>
#include <thread>
#include <functional>

const int TASK_MAX_THRESHOLD = INT32_MAX;
const int THREAD_MAX_THRESHOLD = 100;
const int THREAD_MAX_IDLE_TIME = 60; // 单位：s

// 线程池构造
ThreadPool::ThreadPool()
	: initThreadSize_(0)
	, curThreadSize_(0)
	, idleThreadSize_(0)
	, taskSize_(0)
	, taskQueMaxThresHold_(TASK_MAX_THRESHOLD)
	, poolMode_(PoolMode::MODE_FIXED)
	, isPoolRunning_(false)
	, threadSizeThresHold_(THREAD_MAX_THRESHOLD)
{}

// 线程池析构
ThreadPool::~ThreadPool() 
{
	isPoolRunning_ = false;

	// 等待线程池中所有的线程返回 -> 有两种状态：阻塞 & 正在执行任务中
	std::unique_lock<std::mutex> lock(taskQueMtx_);
	notEmpty_.notify_all();
	exitCond_.wait(lock, [&]()->bool { return threads_.size() == 0; });
}

// 支持用户设置线程池的工作模式
void ThreadPool::setMode(PoolMode mode)
{
	// 运行过程无法修改线程池状态
	if (checkRunningState())
		return;

	poolMode_ = mode;
}

// 支持用户设置任务队列上限阈值
void ThreadPool::setTaskQueMaxThresHold(int threshold)
{
	// 运行过程无法修改线程池状态
	if (checkRunningState())
		return;

	taskQueMaxThresHold_ = threshold;
}

// 支持用户设置线程池Cached模式的线程上限阈值
void ThreadPool::setThreadSizeThresHold(int threshold)
{
	// 运行过程无法修改线程池状态
	if (checkRunningState())
		return;

	if (poolMode_ == PoolMode::MODE_CACHED)
	{
		threadSizeThresHold_ = threshold;
	}
}

// 支持用户向线程池提交任务 -> 用户传入任务对象
Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
	// 获取锁
	std::unique_lock<std::mutex>lock(taskQueMtx_);

	// 线程通信 -> 等待任务队列有空余 wait wait_for 和 wait_until
	// while (taskQue_.size() == taskQueMaxThresHold_) { notFull_.wait(lock); } // 等价于底下
	// notFull_.wait(lock, [&]()->bool { return taskQue_.size() < taskQueMaxThresHold_; }); // 等待状态
	
	// 用户提交任务，最长不能阻塞超过1s，否则判断提交任务失败，返回
	if (!notFull_.wait_for(lock, std::chrono::seconds(1), [&]()->bool { return taskQue_.size() < (size_t)taskQueMaxThresHold_; }))
	{
		// wait_for返回为false，说明用户提交任务时最长阻塞超过了1s
		std::cerr << "Task queue is full, submit task fail." << std::endl;
		// return task->getResult(); // 线程执行完task，task对象就被析构掉了
		return Result(sp, false);
	}

	// 若有空余，将任务放于任务队列中
	taskQue_.emplace(sp);
	taskSize_++;

	// 新放任务后，即可在notEmpty_上进行通知 -> 方便“消费者”消费
	notEmpty_.notify_all();

	// cached模式任务处理比较紧急（适合小而快的任务），需要根据任务数量和空闲线程的数量，判断是否需要创建新的线程
	if (poolMode_ == PoolMode::MODE_CACHED
		&& taskSize_ > idleThreadSize_
		&& curThreadSize_ < threadSizeThresHold_)
	{
		std::cout << ">>> create new thread." << std::endl;

		// 创建新线程
		auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
		// threads_.emplace_back(std::move(ptr));
		int threadId = ptr->getId();
		threads_.emplace(threadId, std::move(ptr));
		// 启动线程
		threads_[threadId]->start();
		// 修改线程数量相关变量
		curThreadSize_++;
		idleThreadSize_++;
	}

	// 返回任务的Result对象
	// return task->getResult();
	return Result(sp);
}

// 开启线程池
void ThreadPool::start(int initThreadSize)
{
	// 设置线程池的运行状态
	isPoolRunning_ = true;

	// 记录初始线程个数
	initThreadSize_ = initThreadSize;
	curThreadSize_ = initThreadSize;
	
	// 创建线程对象
	for (int i = 0; i < initThreadSize_; i++)
	{
		// 创建thread线程对象时，通过绑定器把线程函数给到thread线程对象
		auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
		// threads_.emplace_back(std::move(ptr));
		int threadId = ptr->getId();
		threads_.emplace(threadId, std::move(ptr));
	}

	// 启动所有线程 → 需要执行线程函数
	for (int i = 0; i < initThreadSize_; i++)
	{
		// threads_[i]->start(); // 需要执行一个线程函数
		threads_[i]->start(); // 需要执行一个线程函数
		idleThreadSize_++; // 记录初始空闲线程的数量
	}
}

// 定义线程函数 → 定义于ThreadPool类中，方便访问私有成员 → 再通过绑定器把线程函数给thread线程对象
void ThreadPool::threadFunc(int threadId)
{
	/*std::cout << "begin threadFunc: " << std::this_thread::get_id() << std::endl;
	std::cout << "end threadFunc" << std::this_thread::get_id() << std::endl;*/
	auto lastTime = std::chrono::high_resolution_clock().now();

	// 所有任务必须执行完成，线程池才可以回收所有线程资源
	for(;;)
	{
		std::shared_ptr<Task> task;
		{
			// 获取锁
			std::unique_lock<std::mutex>lock(taskQueMtx_);

			std::cout << "tid: " << std::this_thread::get_id() << "尝试获取任务..." << std::endl;

			// cahced模式下，可能已经创建了很多线程，但空闲时间超过60s，应该回收多余的线程
			
			// 每秒返回一次 -> 如何区分超时返回 or 有任务待执行返回
			while (taskQue_.size() == 0)
			{
				// 线程池结束，回收线程资源
				if (!isPoolRunning_)
				{
					threads_.erase(threadId);
					exitCond_.notify_all();
					std::cout << "threadid:" << std::this_thread::get_id() << "exit!" << std::endl;
					return; // 线程函数结束，线程退出
				}

				if (poolMode_ == PoolMode::MODE_CACHED)
				{
					// 条件变量超时返回
					if (std::cv_status::timeout == notEmpty_.wait_for(lock, std::chrono::seconds(1)))
					{
						auto now = std::chrono::high_resolution_clock().now();
						auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
						if (dur.count() >= THREAD_MAX_IDLE_TIME && curThreadSize_ > initThreadSize_)
						{
							// 开始回收当前线程
							// 修改线程数量相关的值
							// 将线程对象从线程列表容器中删除（threadID - 线程对象）
							threads_.erase(threadId);
							curThreadSize_--;
							idleThreadSize_--;

							std::cout << "threadid:" << std::this_thread::get_id() << "exit!" << std::endl;

							return;
						}
					}
				}
				else
				{
					// 等待notEmpty_条件
					notEmpty_.wait(lock);
				}

				// 判断线程池结束，回收线程资源
				if (!isPoolRunning_)
				{
					threads_.erase(threadId);
					std::cout << "threadid:" << std::this_thread::get_id() << "exit!" << std::endl;
					exitCond_.notify_all();
					return; // 结束线程函数，即结束当前线程
				}
			}

			idleThreadSize_--;

			std::cout << "tid: " << std::this_thread::get_id() << "获取任务成功..." << std::endl;

			// 从任务队列中取一个任务
			task = taskQue_.front();
			taskQue_.pop();
			taskSize_--;

			// 若仍有剩余任务，继续通知其它线程执行任务
			if (taskQue_.size() > 0)
			{
				notEmpty_.notify_all(); // 通知阻塞在notEmpty信号量上的线程
			}

			// 取出一个任务，进行通知
			notFull_.notify_all(); // 通知阻塞在notFull信号量上的线程

		} // unique_lock出了作用域即可释放锁 -> 防止线程在执行任务过程中一直“占用”锁

		// 当前线程负责执行任务
		if (task != nullptr) 
		{
			// task->run();
			task->exec(); // 执行任务完成后将任务返回值通过setVal()赋给Result
		}
		idleThreadSize_++;
		auto lastTime = std::chrono::high_resolution_clock().now(); // 更新线程执行完任务的时间
	}

}

bool ThreadPool::checkRunningState() const
{
	return isPoolRunning_;
}

// ---线程方法实现---
int Thread::generateId_ = 0;
// 线程构造
Thread::Thread(ThreadFunc func)
	: func_(func)
	, threadId_(generateId_++)
{}

// 线程析构
Thread::~Thread() {}


// 启动线程
void Thread::start()
{
	// 创建一个线程来执行一个线程函数
	std::thread t(func_, threadId_); // C++11 线程对象t 和线程函数func_
	t.detach(); // 设置分离线程 → 防止线程对象出作用域自动析构
}

int Thread::getId() const
{
	return threadId_;
}

// ---Task方法实现---
Task::Task()
	: result_(nullptr)
{}

void Task::exec()
{
	if (result_ != nullptr)
	{
		result_->setVal(run()); // 这里发生多态调用 (run)
	}
}

void Task::setResult(Result* res)
{
	result_ = res;
}

// ---Result方法实现---
Result::Result(std::shared_ptr<Task> task, bool isValid)
	: isValid_(isValid)
	, task_(task)
{
	task_->setResult(this);
}

Any Result::get()
{
	if (!isValid_)
	{
		return "";
	}

	sem_.wait(); // task任务如果未执行完，此处会阻塞用户的线程
	return std::move(any_);
}

void Result::setVal(Any any)
{
	// 存储task的返回值
	this->any_ = std::move(any); // 资源转移
	sem_.post(); // 已经获取任务的返回值，增加信号量资源
}